package dkbta;

import static dkbta.util.Utils.readDOMDocFromXML;
import static java.lang.System.currentTimeMillis;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintStream;
import java.io.Writer;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import dkbta.processing.UnitPorts;
import dkbta.util.Log;
import dkbta.util.plan.AbstractPlanMaker;
import dkbta.util.plan.PlanMakerFactory;

/**
 * The coordinator. Its responsibility is to coordinate the start of all computational
 * units, to receive signals upon the shut down of each one and to output relevant info to
 * the log in the process.
 * 
 * @author Uri Kanonov
 */
public class Coordinator {
	private static boolean UNITS_PRESENT;
	private static boolean FILE_INPUT;
	
	public static void main(String[] args) throws IOException{
		Log log = Log.init("Coordinator");
		try{
			// Loading the coordinator's "plan"
			FILE_INPUT = CoordinatorConfig.isFileInput();
			UNITS_PRESENT = CoordinatorConfig.isUnits();
			
			Map<Integer, UnitPorts> unitPorts = CoordinatorConfig.getUnits();
			int coordinatorPort = CoordinatorConfig.getCoordinatorServerAddress().getPort();
			int numUnits = unitPorts.size();
			
			String taskFileName = CoordinatorConfig.getTaskFileName();
			if(!FILE_INPUT){//getting task from Task Server and writing it into task file
				
				InetSocketAddress address = CoordinatorConfig.getTaskServerAddress();
				Socket socket = new Socket(address.getHostName(), address.getPort());
				BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
				FileOutputStream fos = new FileOutputStream(taskFileName);
				String strInput;
				
				while ((strInput = in.readLine()) != null) {
					new PrintStream(fos).println(strInput);
				}
				fos.close();
			}
			
			File planFile = new File(taskFileName);
			AbstractPlanMaker planMaker = PlanMakerFactory.getPlanMaker(CoordinatorConfig.getPlanMakerName(), readDOMDocFromXML(planFile).getDocumentElement());
			planMaker.makePlans(taskFileName, numUnits);
						
			if(UNITS_PRESENT){
				CountDownLatch latch = new CountDownLatch(numUnits);
						
				// Preparing and starting the server that will listen to shut down signals
				// from the other units. We do this before even starting the units so that
				// if the start-up process takes a long time and a certain unit finishes very
				// quickly, we won't miss its shut down signal
				startCoordinatorServerThread(log, coordinatorPort, latch);
				// Sending start signals to all units
				long startTime = currentTimeMillis(); // Noting the overall start time
				startAllUnits(log, unitPorts, numUnits, latch);
	
				// Waiting for all units to finish
				latch.await();
				long endTime = currentTimeMillis(); // Noting the overall end time
	
				long millis = endTime - startTime;
				long hours = millis / (1000 * 60 * 60);
				millis -= hours * (1000 * 60 * 60);
				long minutes = millis / (1000 * 60);
				millis -= minutes * (1000 * 60);
				long seconds = millis / (1000);
				millis -= seconds * (1000);
	
				log.info("All " + numUnits + " units have finished\n" + "Total work time: "
						+ hours + " hours, " + minutes + " minutes, " + seconds
						+ " seconds, " + millis + " milliseconds");
			}
		}catch(Throwable t){
			log.severe("Coordinator crashed", t);
		}
	}

	private static void startAllUnits(Log log, Map<Integer, UnitPorts> unitPorts,
		int numUnits, CountDownLatch latch){
		log.info("Attempting to start " + numUnits + " units");
		for (Map.Entry<Integer, UnitPorts> unit : unitPorts.entrySet()){
			try{
				Socket socket = new Socket();
				socket.connect(unit.getValue().controlPort);
				Writer osw = new OutputStreamWriter(socket.getOutputStream());
				osw.write("Start");
				osw.flush();
				try{
					socket.getInputStream().read();
				}catch(Throwable t){}
			}catch(Exception e){
				log.severe("Unable to send start signal to unit #" + unit.getKey(), e);
				// Decrementing the count so that the coordinator won't wait for the
				// unit that could not be started
				latch.countDown();
			}
		}
		log.info("Succesfully started and waiting for " + latch.getCount() + " units");
	}

	private static void startCoordinatorServerThread(final Log log,
		final int coordinatorPort, final CountDownLatch latch){
		// Preparing the server
		Thread server = new Thread("Coordinator server thread"){
			@Override
			public void run(){
				try{
					ServerSocket srv = new ServerSocket(coordinatorPort);
					while (true){
						// Waiting for a connection
						final Socket socket = srv.accept();

						// We received a connection (shut down signal)
						// from a unit. We process it on another thread.
						new Thread(){
							@Override
							public void run(){
								// Printing info to the log
								long toGo = latch.getCount() - 1;
								try{
									String unitId = new BufferedReader(
											new InputStreamReader(socket.getInputStream()))
											.readLine();
									log.info("Unit #" + unitId + " has finished, " + toGo
											+ " to go");
								}catch(IOException e){
									log.warning("A unit has finished however "
											+ "cannot read its number, " + toGo
											+ " to go");
								}
								// Noting that a unit has finished
								latch.countDown();
							}
						}.start();
					}
				}catch(Throwable t){
					log.severe("Coordinator server thread crashed", t);
					System.exit(1);

				}

			}
		};
		// Setting the server as a daemon so it won't interfere with shut down
		server.setDaemon(true);
		server.start(); // Starting the server
	}
}
